package com.myflink.day05;

import com.myflink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import javax.annotation.Nullable;

/**
 * @author Shelly An
 * @create 2020/9/24 8:48
 * 将水位值高于5cm的值输出到side output
 */
public class ProcessFunction_SideOutPut {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //使用侧输出流
        //1. 定义一个OutputTag，给定一个名称
        //2. 使用上下文对象 ctx.output(outputtag对象,放入侧输出流的数据)
        //3. 获取侧输出流 => DataStream.getSideOutput(outputtag对象）
        OutputTag<String> outputTag = new OutputTag<String>("vc alarm");

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 4444)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
                    }
                })
                .assignTimestampsAndWatermarks(
                        new AssignerWithPunctuatedWatermarks<WaterSensor>() {
                            private Long maxTs = Long.MIN_VALUE;

                            @Nullable
                            @Override
                            public Watermark checkAndGetNextWatermark(WaterSensor lastElement, long extractedTimestamp) {
                                maxTs = Math.max(maxTs, extractedTimestamp);
                                return new Watermark(maxTs);
                            }

                            @Override
                            public long extractTimestamp(WaterSensor element, long previousElementTimestamp) {
                                return element.getTs() * 1000L;
                            }
                        }
                );

        SingleOutputStreamOperator<WaterSensor> resultDS = sensorDS
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {


                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                        if (value.getVc()>5) {
                            //水位高于阈值，用侧输出流告警
                            ctx.output(outputTag,"水位高于阈值5！！！");
                        }
                        out.collect(value);
                    }
                });

        resultDS.print("result");
        resultDS.getSideOutput(outputTag).print("vc alarm:");

        env.execute();
    }
}
